package com.culture.bootdemo.utils;

import com.google.common.collect.Sets;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

/**
 * 读取大数据分布式文件 HDFS文件
 * ===高可用版本：
 */
public class HDFSUtil {
    private static int THREAD_NUM = 10;


    private static String ip1 = "10.7.29.51:8020";
    private static String ip2 = "10.7.29.52:8020";

    private static String NAME_SERVICE_ID = "seCluster1";
    private static FileSystem FS;

    static {
        try {
            FS = getHdfs(NAME_SERVICE_ID, ip1, ip2);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        try {
            Set<Object> set = Sets.newConcurrentHashSet();
            List<String> content = HDFSUtil.readFoldersByCallBack("/user/algo/push/offline/pair_result/2021-07-01/1000189");
            String temp = "";
            int cols = 9;
            List<String> downLoadPid = new ArrayList<>();
            for (String s : content) {
                String lineString = temp + s;
                String[] lines = lineString.split("\n");
                int lastIndex = lines.length - 1;
                if (lines[lastIndex].split("\t").length < cols) {
                    temp = lines[lastIndex];
                } else {
                    for (String line : lines) {
                        String[] lineStr = line.split("\t");
                        String custId = lineStr[0];
                        String taskId = lineStr[8];
                    }
                }
            }

            System.out.println("总个数：" + downLoadPid.size());
        } catch (IOException e) {
            e.printStackTrace();
        }


    }

    private static FileSystem getHdfs(String nameServiceId, String ip1, String ip2) throws IOException {
        Configuration config = new Configuration();
        config.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        config.setBoolean("fs.hdfs.impl.disable.cache", true);
        config.set("fs.defaultFS", "hdfs://" + nameServiceId);
        config.set("dfs.nameservices", nameServiceId);
        config.set("dfs.ha.namenodes." + nameServiceId, "nn1,nn2");
        config.set("dfs.namenode.rpc-address." + nameServiceId + ".nn1", ip1); // 10.7.29.51
        config.set("dfs.namenode.rpc-address." + nameServiceId + ".nn2", ip2); // 10.7.29.52
        config.set("dfs.client.failover.proxy.provider." + nameServiceId, "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

        if (FS == null) {
            FS = FileSystem.get(URI.create("hdfs://" + nameServiceId), config);
        } else {
            return FS;
        }
        return FS;
    }

    /**
     * 解析文件目录
     */
    public static List<String> parsePath(String path) throws IOException {
        List<String> res = new ArrayList<String>();
        FileStatus[] fileStatuses = FS.globStatus(new Path(path));
        for (FileStatus fileStatus : fileStatuses) {
            res.add(fileStatus.getPath().toString());
        }
        return res;
    }

    /**
     * 获取文件目录
     */

    public static List<String> readFoldersByCallBack(String path) throws IOException {
        List<String> holder = parsePath(path);
        List<String> result = new ArrayList<String>();
        for (int i = 0; i < holder.size(); i++) {
            List<String> all_children = new ArrayList<String>();
            Long parsePathStart = System.currentTimeMillis();
            listChildren(holder.get(i), all_children);
            Long parsePathEnd = System.currentTimeMillis();
            System.out.println("子目录解析耗时：" + (parsePathEnd - parsePathStart) + "ms");
            Long getResultStart = System.currentTimeMillis();
            ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUM);
            List<FutureTask<List<String>>> tasks = new ArrayList<FutureTask<List<String>>>();
            for (final String single_file : all_children) {
                FutureTask<List<String>> future = new FutureTask<>(() -> readSingleFile(new Path(single_file)));
                tasks.add(future);
            }
            for (FutureTask<List<String>> task : tasks) {
                executor.execute(task);
            }
            for (FutureTask<List<String>> task : tasks) {
                try {
                    List<String> list = task.get();
                    result.addAll(list);
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
            executor.shutdown();
            Long getResultEnd = System.currentTimeMillis();
            System.out.println("读取数据耗时：" + (getResultEnd - getResultStart) + "ms");
        }
        return result;
    }

    /**
     * 读取单个文件
     */
    public static List<String> readSingleFile(Path path) throws IOException {
        List<String> result = new ArrayList<String>();
        if (FS.exists(path)) {
            System.out.println(path.toString());
            FSDataInputStream inputStream = FS.open(path);
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            IOUtils.copy(inputStream, outputStream);
            result.add(outputStream.toString());
            inputStream.close();
            outputStream.close();
        }
        return result;
    }

    /**
     * 获取子目录
     */
    public static void listChildren(String fullName, List<String> holder) throws IOException {
        String regrex;
        if (FS.isDirectory(new Path(fullName))) {
            regrex = "/*";
        } else {
            regrex = "";
        }
        FileStatus[] fileStatuses = FS.globStatus(new Path(fullName + regrex));
        if (fileStatuses != null) {
            for (FileStatus fileStatus : fileStatuses) {
                Path filePath = fileStatus.getPath();
                if (!filePath.toString().contains("_SUCCESS")) {
                    if (FS.isFile(filePath)) {
                        holder.add(filePath.toString());
                    } else {
                        listChildren(filePath.toString(), holder);
                    }
                }
            }
        }
    }
}
